
/*
 * En este archivo implementamos la funcionalidad del nodo MASTER. El comportamiento del mismo es el siguiente:
 *
 * Los pasos a llevar a cabo por el MASTER son los siguientes:
 *  --> Establecer el tamaño de la sublista que le enviaremos a los workers. A cada nodo se le puede enviar
 *      1 o 2 listas.
 *  --> Efectuamos el envío de las sublistas a los workers.
 *  --> Agregamos a cada nodo a la lista de workers en proceso (una lista por nivel de división de la lista inicial)
 *  --> Mientras existan nodos en proceso:
 *      <> Esperamos a que un worker termine su trabajo.
 *      <> Se saca el worker de la lista de workers en proceso del nivel actual del worker.
 *      <> Si no hay workers en espera en la cola del nivel del worker que ha finalizado su trabajo.
 *          - Se coloca al worker en la cola de espera del mismo nivel de la lista del que fue removido previamente.
 *      <> Sino
 *          - Se remueve un worker en espera de la cola de espera del nivel correspondiente.
 *          - Se le ordena al worker que ha terminado, que envie sus datos al worker que se obtuvo en el paso anterior,
 *          - Se agrega el worker al que se le han enviado los datos, a la lista de workers en proceso del nivel
 *              inmediato superior.
 *  --> Fin Mientras
 *  --> Esperamos a que el único nodo restante envíe la lista ordenada al MASTER.
 */

#include "../common/vectores.h"
#include "../common/entradasalida.h"

#include "master.h"
#include "listaworkers.h"
#include "mensajes.h"

#include <stdio.h>
#include <malloc.h>
#include <mpi.h>
#include <string.h>
#include <math.h>


/*
 * Ordena el vector pasado por parametro utilizando el algoritmo de mergesort,
 * en forma paralela.
 */
struct VECTOR *ordenarVectorEnParalelo(struct VECTOR *vectorDesordenado);
struct LISTA_WORKERS *obtenerTodosLosWorkers();
void distribuirVectorDesordenado(struct VECTOR *vector, struct LISTA_WORKERS *todosLosWorkers, struct LISTA_WORKERS *workersEnProceso);
int existenNodosEnProceso(struct LISTA_WORKERS *workersEnEspera, struct LISTA_WORKERS *workersEnProceso);
int esperarPedidoDeWorker();
struct WORKER *seleccionarWorker(struct LISTA_WORKERS *workersEnEspera);
struct WORKER **obtenerOrigenYDestino(struct WORKER *worker1, struct WORKER *worker2);
void ordenarEnvioDeVector(struct WORKER *workerQueEnvia, struct WORKER *workerQueRecibe);
void ordenarRecepcionDeVector(struct WORKER *workerQueRecibe, struct WORKER *workerQueEnvia);
int *obtenerResultadoDelUltimoWorker();

int timestamps[10000];
int accion[10000];
int parametros[10000];
int contadorAcciones = 0;
int inicial;

#define INICIALIZANDO       0
#define PROCESANDO          1
#define ESPERANDO_PEDIDOS   2
#define MENSAJE_FIN         3
#define MENSAJE_ORDENAR     4
#define MENSAJE_MERGEAR     5
#define MENSAJE_ENVIAR      6
#define ESPERANDO_ULTIMO    7
#define RECIBIENDO_ULTIMO   8
#define FINALIZANDO         9

void registrarEvento(int numeroEvento, int parametro) {
    timestamps[contadorAcciones] = MPI_Wtime();
    accion[contadorAcciones] = numeroEvento;
    parametros[contadorAcciones] = parametro;
    contadorAcciones++;
}

void escribirDump() {
    char *nombre = malloc(sizeof(char) * 20);
    int rank;
    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
    sprintf(nombre, "debug-%d.csv", rank);
    FILE *filePointer = fopen(nombre, "w");
    int i;
    for(i=0;i<contadorAcciones;i++) {

        char *accionName;
        switch (accion[i]) {

            case 0: accionName = "INICIALIZANDO";
                break;
            case 1: accionName = "PROCESANDO";
                break;
            case 2: accionName = "ESPERANDO_PEDIDOS";
                break;
            case 3: accionName = "MENSAJE_FIN";
                break;
            case 4: accionName = "MENSAJE_ORDENAR";
                break;
            case 5: accionName = "MENSAJE_MERGEAR";
                break;
            case 6: accionName = "MENSAJE_ENVIAR";
                break;
            case 7: accionName = "ESPERANDO_ULTIMO";
                break;
            case 8: accionName = "RECIBIENDO_ULTIMO";
                break;
            case 9: accionName = "FINALIZANDO";
                break;
        }

        int tiempo;
        if (i==0) {
            tiempo = timestamps[i]-inicial;
        } else {
            tiempo = timestamps[i] - timestamps[i-1];
        }
        fprintf(filePointer, "%d,%d,%s,%d\n", tiempo, timestamps[i] - inicial, accionName, parametros[i]);
    }
    fclose(filePointer);
}

/*
 * Implementacion de función definida en cabecera
 */
void master() {
    inicial = MPI_Wtime();
    registrarEvento(INICIALIZANDO, -1);

    // Imprimimos la bienvenida
    printf("============================================================================\n");
    printf("Ordenamiento de números mediante Merge Sort. VERSIÓN PARALELA\n");
    printf("============================================================================\n");

    struct VECTOR *vectorInicial; // Vector a ordenar
    struct VECTOR *vectorOrdenado; // Vector ordenado
    double instanteInicial; // Instante inicial del calculo
    double instanteFinal; // Instante final del calculo

    // Obtenemos el vector a ordenar, registramos el instante inicial, y ordenamos el vector
    vectorInicial = obtenerVectorDelUsuario();

    registrarEvento(PROCESANDO, -1);

    instanteInicial = MPI_Wtime();
    vectorOrdenado = ordenarVectorEnParalelo(vectorInicial);

    // Liberamos la memoria reservada
    eliminarVector(vectorInicial);

    // Calculamos e imprimimos el tiempo que tardó en ejecutarse el programa
    instanteFinal = MPI_Wtime();
    int tiempoEjecucion = instanteFinal - instanteInicial;
    printf("\nTiempo de ejecucion: %d segundos\n", tiempoEjecucion);

    // Guardamos el vector ordenado
    almacenarVector(vectorOrdenado, "vector ordenado");

    // Imprimimos la despedida
    printf("\n====================================================================\n");
    printf("Fin del programa\n");
    printf("====================================================================\n\n");

    escribirDump();

}


struct VECTOR *ordenarVectorEnParalelo(struct VECTOR *vectorDesordenado) {

    // Necesitamos una referencia a todos los workers disponibles
    struct LISTA_WORKERS *todosLosWorkers = obtenerTodosLosWorkers();

    // Necesitamos saber que workers estan trabajando y cuales esperando una tarea
    struct LISTA_WORKERS *workersEnEspera = inicializarListaDeWorkers(todosLosWorkers->cantidad);
    struct LISTA_WORKERS *workersEnProceso = inicializarListaDeWorkers(todosLosWorkers->cantidad);

    distribuirVectorDesordenado(vectorDesordenado, todosLosWorkers, workersEnProceso);

    while (existenNodosEnProceso(workersEnEspera, workersEnProceso)) {

        // Esperamos que algún worker termine
        int idWorkerOrigen = esperarPedidoDeWorker();

        // Sacamos al worker que termino, de la lista de workers en proceso
        struct WORKER *workerOrigen = quitarWorkerDeLista(workersEnProceso, idWorkerOrigen);

        if (workersEnEspera->cantidad == 0) { // Agregamos el nodo origen a la lista de workers en espera
            colocarWorkerEnLista(workersEnEspera, workerOrigen);
        } else {
            
            // Obtenemos el worker que originó el pedido, se enviarán sus listas
            struct WORKER *workerSeleccionado = seleccionarWorker(workersEnEspera);
            quitarWorkerDeLista(workersEnEspera, workerSeleccionado->idWorker);

            // Decidimos que worker le envia los datos al otro worker
            struct WORKER **datosEnvio = obtenerOrigenYDestino(workerOrigen, workerSeleccionado);
            struct WORKER *workerQueEnvia = datosEnvio[0];
            struct WORKER *workerQueRecibe = datosEnvio[1];
            free(datosEnvio);

            // Enviamos las ordenes a los workers de que envien o reciban del otro worker
            ordenarEnvioDeVector(workerQueEnvia, workerQueRecibe);
            ordenarRecepcionDeVector(workerQueRecibe, workerQueEnvia);

            // Colocamos al worker destino en la lista de workers en proceso y actualizamos su tam de vector
            workerQueRecibe->tamVector += workerQueEnvia->tamVector;
            colocarWorkerEnLista(workersEnProceso, workerQueRecibe);
        }
    }

    // Esperamos que el ultimo worker nos envíe su lista, creamos el objeto vector y lo devolvemos
    struct VECTOR *vectorOrdenado = crearVectorSinElementos(vectorDesordenado->dimension);
    vectorOrdenado->elementos = obtenerResultadoDelUltimoWorker();

    registrarEvento(FINALIZANDO, -1);

    // Liberamos toda la memoria que ocupamos
    eliminarListaDeWorkers(todosLosWorkers, 1);
    eliminarListaDeWorkers(workersEnEspera, 0);
    eliminarListaDeWorkers(workersEnProceso, 0);

    return vectorOrdenado;

}


/*
 * Genera una lista de todos los workers disponibles en el cluster.
 * Los workers de la lista no tienen datos.
 */
struct LISTA_WORKERS *obtenerTodosLosWorkers() {

    // Obtenemos la cantidad total de WORKERs
    int cantidadNodos;
    MPI_Comm_size(MPI_COMM_WORLD, &cantidadNodos);
    int cantidadWorkers = cantidadNodos - 1;

    // Creamos la lista de workers
    struct LISTA_WORKERS *listaWorkers = inicializarListaDeWorkers(cantidadWorkers);

    // Creamos los workers uno a uno
    int i;
    for (i=0; i<cantidadWorkers; i++) {
        struct WORKER *nuevoWorker = malloc(sizeof(struct WORKER));
        nuevoWorker->idWorker = i+1;
        nuevoWorker->tamVector = 0;
        colocarWorkerEnLista(listaWorkers, nuevoWorker);
    }

    return listaWorkers;
}


/*
 * Distribuye el vector pasado por parametro, a los workers de la lista todosLosWorkers.
 * Coloca todos los workers en la lista workersEnProceso, tambien les asigna los datos correspondientes.
 */
void distribuirVectorDesordenado(struct VECTOR *vector, struct LISTA_WORKERS *todosLosWorkers, struct LISTA_WORKERS *workersEnProceso) {

    int tamVector = vector->dimension;

    // Decidimos y asignamos cuantos elementos enviaremos a cada worker
    int cantBase = tamVector / todosLosWorkers->cantidad;
    int cantSobrante = tamVector % todosLosWorkers->cantidad;
    int particiones = 1;
    
    // Ajustamos la cantidad base, de acuerdo al tamaño máximo de mensaje
    if (cantBase > TAM_MAX_PAQUETE) {
        particiones = ceil((cantBase * 1.0) / TAM_MAX_PAQUETE;
    }

    int i;
    for (i=0; i<cantSobrante; i++) {
        todosLosWorkers->workers[i]->tamVector = cantBase + 1;
    }

    int j;
    for (j=i; j<todosLosWorkers->cantidad; j++) {
        todosLosWorkers->workers[j]->tamVector = cantBase;
    }

    // Le enviamos a cada nodo el vector que debe procesar y lo asignamos a la lista de workers en proceso
    int *contVector = vector->elementos;
    int offsetEnvioInicial = 0;
    
    int p;
    for (p=0; p<todosLosWorkers->cantidad; p++) {

        // Obtenemos el receptor actual
        struct WORKER *receptor = todosLosWorkers->workers[p];

        // Si el worker posee una cantidad de 0 elementos, entonces debemos enviarle un mensaje de finalizacion
        MPI_Request *request = malloc(sizeof(MPI_Request));
        
        if (receptor->tamVector == 0) {
            int dummyData = 0;
            registrarEvento(MENSAJE_FIN, receptor->idWorker);
            MPI_Isend(&dummyData, 1, MPI_INT, receptor->idWorker, FINALIZAR, MPI_COMM_WORLD, request);
            registrarEvento(PROCESANDO, -1);
        } else {

            int envioBase = receptor->tamVector / particiones;
            int envioRestante = receptor->tamVector % particiones;
            int *tams = malloc(sizeof(int) * particiones);

            int n;
            for(n=0;n<particiones;n++) {
                tams[n] = envioBase;
            }
            for(n=0;n<envioRestante;n++) {
                tams[n] += 1;
            }

            int w;
            for (w=0; w<particiones; w++) {

                int mensaje;
                if (w == particiones - 1) {
                    mensaje = ORDENAR_Y_JUNTAR;
                } else {
                    mensaje = ORDENAR_NUEVA_LISTA;
                }

                int *offset = contVector + offsetEnvioInicial;
                registrarEvento(MENSAJE_ORDENAR, receptor->idWorker);
                MPI_Isend(offset, tams[w], MPI_INT, receptor->idWorker, mensaje, MPI_COMM_WORLD, request);

                registrarEvento(PROCESANDO, -1);

                // Actualizamos el desplazamiento y colocamos el worker en al lista de qorkers en proceso
                offsetEnvioInicial += tams[w];
            }

            free(tams);
            colocarWorkerEnLista(workersEnProceso, receptor);
            
        }
    }
}


/*
 * Devuelve falso solo si la lista de workers en espera esta vacía y queda mas de un worker en proceso.
 */
int existenNodosEnProceso(struct LISTA_WORKERS *workersEnEspera, struct LISTA_WORKERS *workersEnProceso) {
    return ! (workersEnEspera->cantidad == 0 && workersEnProceso->cantidad == 1);
}


/*
 * Se queda esperando que un worker le envie una solucitud al master.
 * Cuando detecta una solicitud, devuelve el id del nodo que requiere una tarea.
 */
int esperarPedidoDeWorker() {

    // Esperamos que algun WORKER nos envie un mensaje de espera
    MPI_Status mensaje;
    int dummyBuffer;
    registrarEvento(ESPERANDO_PEDIDOS, -1);
    MPI_Recv(&dummyBuffer, 1, MPI_INT, MPI_ANY_SOURCE, ESPERANDO_ORDENES, MPI_COMM_WORLD, &mensaje);
    registrarEvento(PROCESANDO, mensaje.MPI_SOURCE);
    // Ya recibimos un pedido de tareas por parte de un WORKER, devolvemos el id del worker
    return mensaje.MPI_SOURCE;

}


/*
 * Selecciona el worker en espera con el vector mas chico de todos, dentro de la lista de workers en espera.
 * El worker seleccionado no es removido de la lista.
 */
struct WORKER *seleccionarWorker(struct LISTA_WORKERS *workersEnEspera) {

    int menorTamanio = 1000000000;
    int indiceMenorTamanio = -1;

    // Recorremos la lista completa y guardamos la referencia al mas pequeño
    int i;
    for (i=0; i<workersEnEspera->capacidadMaxima; i++) {
        struct WORKER *worker = workersEnEspera->workers[i];
        if (worker != NO_WORKER && worker->tamVector < menorTamanio) {
            indiceMenorTamanio = i;
            menorTamanio = worker->tamVector;
        }
    }

    return workersEnEspera->workers[indiceMenorTamanio];
}


/*
 * Devuelve un vector de 2 workers, siendo el primero el origen de la transmisión
 * y el segundo el receptor.
 */
struct WORKER **obtenerOrigenYDestino(struct WORKER *worker1, struct WORKER *worker2) {

    // Creamos el array a devolver
    struct WORKER **origenDestino = malloc(sizeof(struct WORKER *) * 2);

    // El origen es el que tengo el vector mas pequeño
    if (worker1->tamVector < worker2->tamVector) {
        origenDestino[0] = worker1;
        origenDestino[1] = worker2;
    } else {
        origenDestino[0] = worker2;
        origenDestino[1] = worker1;
    }

    return origenDestino;
}


/*
 * Le ordena al worker que debe enviar, que envie su vector al worker receptor.
 */
void ordenarEnvioDeVector(struct WORKER *workerQueEnvia, struct WORKER *workerQueRecibe) {

    // Obtenemos el id del nodo que debera enviar su lista
    // Obtenemos el id del nodo que deberá recibir la lista enviada por el transmisor
    int idTransmisor = workerQueEnvia->idWorker;
    int *idReceptor = &workerQueRecibe->idWorker;

    // Enviamos la orden en forma asincronica
    MPI_Request *request = malloc(sizeof(MPI_Request));
    registrarEvento(MENSAJE_ENVIAR, idTransmisor);
    MPI_Isend(idReceptor, 1, MPI_INT, idTransmisor, ENVIAR_LISTA, MPI_COMM_WORLD, request);
    registrarEvento(PROCESANDO, -1);

}


/*
 * Le ordena al worker receptor que reciba el vector del worker emisor.
 */
void ordenarRecepcionDeVector(struct WORKER *workerQueRecibe, struct WORKER *workerQueEnvia) {
    
    // Obtenemos el id del nodo que debera enviar su lista
    // Obtenemos el id del nodo que deberá recibir la lista enviada por el transmisor
    int *idTransmisor = &workerQueEnvia->idWorker;
    int idReceptor = workerQueRecibe->idWorker;

    // Enviamos la orden en forma asincronica
    MPI_Request *request = malloc(sizeof(MPI_Request));
    registrarEvento(MENSAJE_MERGEAR,idReceptor);
    MPI_Isend(idTransmisor, 1, MPI_INT, idReceptor, RECIBIR_LISTA, MPI_COMM_WORLD, request);
    registrarEvento(PROCESANDO,-1);
}


/*
 * Obtiene el vector ordenado del ultimo worker activo del cluster.
 */
int *obtenerResultadoDelUltimoWorker() {

    // Esperamos a que el ultimo worker nos pida una nueva tarea
    int idUltimoWorker = esperarPedidoDeWorker();

    // Le solicitamos que nos envíe su lista, en forma sincronica
    int idMaster = 0;
    MPI_Send(&idMaster, 1, MPI_INT, idUltimoWorker, ENVIAR_LISTA, MPI_COMM_WORLD);

    // Nos quedamos esperando a que nos envíe su vector
    MPI_Status estadoMensaje;
    registrarEvento(ESPERANDO_ULTIMO, -1);
    MPI_Probe(idUltimoWorker, TRANSFERIR_LISTA, MPI_COMM_WORLD, &estadoMensaje);
    registrarEvento(PROCESANDO, idUltimoWorker);

    // Obtenemos el tamaño del vector que nos envía el ultimo worker y creamos el buffer necesario para recibirlo
    int tamVectorRemoto;
    MPI_Get_count(&estadoMensaje, MPI_INT, &tamVectorRemoto);
    int *vectorRemoto = malloc(sizeof(int) * tamVectorRemoto);

    registrarEvento(RECIBIENDO_ULTIMO, idUltimoWorker);
    // Recibimos el vector remoto enviado por el ultimo worker
    MPI_Recv(vectorRemoto, tamVectorRemoto, MPI_INT, idUltimoWorker, TRANSFERIR_LISTA, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
    registrarEvento(PROCESANDO, -1);

    return vectorRemoto;
}



